package org.example;

import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.DeleteRowsEventData;
import com.github.shyiko.mysql.binlog.event.EventData;
import com.github.shyiko.mysql.binlog.event.EventHeaderV4;
import com.github.shyiko.mysql.binlog.event.EventType;
import com.github.shyiko.mysql.binlog.event.TableMapEventData;
import com.github.shyiko.mysql.binlog.event.UpdateRowsEventData;
import com.github.shyiko.mysql.binlog.event.WriteRowsEventData;

import java.io.IOException;
import java.io.Serializable;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * 实践基于binlog监听
 * 代码示例来自：https://zhuanlan.zhihu.com/p/11251195290
 * mysql8版本过高，更改依赖
 *     <dependency>
 *       <groupId>com.zendesk</groupId>
 *       <artifactId>mysql-binlog-connector-java</artifactId>
 *       <version>0.30.1</version>
 *     </dependency>
 */
public class ListenerApp {


    private static final Map<Long, TableMapEventData> tableMap = new HashMap<>();

    public static void main(String[] args) {
        String host = "localhost"; // MySQL 服务器地址
        int port = 3306;           // MySQL 端口
        String username = "root"; // Binlog 订阅用户
        String password = "guochuang123"; // 用户密码

        // 初始化 Binlog 客户端
        BinaryLogClient client = new BinaryLogClient(host, port, username, password);
        client.setServerId(112); // 设置唯一的 serverId

        // 注册事件监听器
        client.registerEventListener(event -> {
            EventHeaderV4 headerV4 = event.getHeader();
            EventType eventType = headerV4.getEventType();
            //每张表有一个表id，对应事件信息（表名，库名等其他信息）
            if (eventType == EventType.TABLE_MAP) {
                TableMapEventData eventData = event.getData();
                tableMap.put(eventData.getTableId(), eventData);
            } else {
                //insert,update,delete事件
                if (EventType.isRowMutation(eventType)) {
                    EventData eventData = event.getData();

                    //update
                    if (eventData instanceof UpdateRowsEventData) {
                        UpdateRowsEventData updateRowsEventData = (UpdateRowsEventData) eventData;
                        long tableId = updateRowsEventData.getTableId();
                        List<Map.Entry<Serializable[], Serializable[]>> rows = updateRowsEventData.getRows();
                        TableMapEventData tableMapEventData = tableMap.get(tableId);

                        if (tableMapEventData != null) {
                            String database = tableMapEventData.getDatabase();
                            String table = tableMapEventData.getTable();
                            if (EventType.isUpdate(eventType)) {
                                update(database, table, rows);
                            }
                        }
                    }

                    //insert
                    if (eventData instanceof WriteRowsEventData) {
                        WriteRowsEventData writeRowsEventData = (WriteRowsEventData) eventData;
                        long tableId = writeRowsEventData.getTableId();
                        List<Serializable[]> rows = writeRowsEventData.getRows();
                        TableMapEventData tableMapEventData = tableMap.get(tableId);
                        if (tableMapEventData != null) {
                            String database = tableMapEventData.getDatabase();
                            String table = tableMapEventData.getTable();
                            if (EventType.isWrite(eventType)) {
                                insert(database, table, rows);
                            }
                        }
                    }

                    //delete
                    if (eventData instanceof DeleteRowsEventData) {
                        DeleteRowsEventData deleteRowsEventData = (DeleteRowsEventData) eventData;
                        long tableId = deleteRowsEventData.getTableId();
                        List<Serializable[]> rows = deleteRowsEventData.getRows();
                        TableMapEventData tableMapEventData = tableMap.get(tableId);
                        if (tableMapEventData != null) {
                            String database = tableMapEventData.getDatabase();
                            String table = tableMapEventData.getTable();
                            if (EventType.isDelete(eventType)) {
                                delete(database, table, rows);
                            }
                        }
                    }
                }
            }
        });

        // 启动 Binlog 客户端
        try {
            client.connect();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    //insert操作
    public static void insert(String databaseName, String tableName, List<Serializable[]> data) {
        List<ColumnMetadata> columns = JdbcUtil.getColumns(databaseName, tableName);
        BinlogEvent binlogEvent = buildBinlogEvent(databaseName, tableName, "insert");
        data.forEach(row -> {
            binlogEvent.setData(ConverEntity.toEntity(columns, row));
        });
        System.out.println("库名：" + databaseName + "，表名：" + tableName + "，数据：" + binlogEvent);
    }

    //update操作
    public static void update(String databaseName, String tableName, List<Map.Entry<Serializable[], Serializable[]>> data) {
        List<ColumnMetadata> columns = JdbcUtil.getColumns(databaseName, tableName);
        BinlogEvent binlogEvent = buildBinlogEvent(databaseName, tableName, "update");
        data.forEach(row -> {
            binlogEvent.setData(ConverEntity.toEntity(columns, row.getValue()));
            binlogEvent.setOriginalData(ConverEntity.toEntity(columns, row.getKey()));
        });
        System.out.println("库名：" + databaseName + "，表名：" + tableName + "，数据：" + binlogEvent);
    }

    //delete操作
    public static void delete(String databaseName, String tableName, List<Serializable[]> data) {
        List<ColumnMetadata> columns = JdbcUtil.getColumns(databaseName, tableName);
        BinlogEvent binlogEvent = buildBinlogEvent(databaseName, tableName, "delete");
        data.forEach(row -> {
            binlogEvent.setData(ConverEntity.toEntity(columns, row));
        });
        System.out.println("库名：" + databaseName + "，表名：" + tableName + "，数据：" + binlogEvent);
    }

    //构建BinlogEvent
    private static BinlogEvent buildBinlogEvent(String databaseName, String tableName, String operationType) {
        BinlogEvent binlogEvent = new BinlogEvent<>();
        binlogEvent.setDatabase(databaseName);
        binlogEvent.setTable(tableName);
        binlogEvent.setOperationType(operationType);
        return binlogEvent;
    }




}